package org.gthelper.hdfs;

import java.io.IOException;
import java.io.OutputStream;
import java.net.URI;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Text;
import org.gthelper.data.SendData;
import org.gthelper.hdpWritable.HdpRenderedImageWritable;

public class HdfsMA {
	
	private static String ndUrl = "hdfs://localhost:9000";
	
	public static void setUrl(String hdfsroot)
	{
		ndUrl=hdfsroot;
	}
	
	public static String  getUrl()
	{
		return ndUrl;
	}
	
	public void saveSeqFile(String path,Text key,HdpRenderedImageWritable val) throws IOException
	{
		Configuration conf = new Configuration();

		FileSystem fs = FileSystem.get(URI.create(ndUrl+path), conf);
		Path hdfsPath = new Path(ndUrl+path);
		@SuppressWarnings("deprecation")
		SequenceFile.Writer writer = SequenceFile.createWriter(fs,conf,hdfsPath,Text.class,HdpRenderedImageWritable.class);
		writer.append(key, val);
//		writer.hflush();
		IOUtils.closeStream(writer);
	}
	
	
	public void readSeqFile(String path,Map<String,HdpRenderedImageWritable> resultMap,List<String> keyList) throws IOException
	{
		Configuration conf = new Configuration();

		FileSystem fs = FileSystem.get(URI.create(ndUrl+path), conf);
		Path hdfsPath = new Path(ndUrl+path);
		@SuppressWarnings({ "deprecation" })
		SequenceFile.Reader reader = new SequenceFile.Reader(fs, hdfsPath, conf);
		Text key = new Text();
		HdpRenderedImageWritable val = new HdpRenderedImageWritable();
		
		while(reader.next(key, val))
		{
			resultMap.put(key.toString(), val);
			keyList.add(key.toString());
		}
		IOUtils.closeStream(reader);
		
	}
	
	public HdpRenderedImageWritable readSingleSeqFile(String path) throws IOException
	{
		Configuration conf = new Configuration();

		FileSystem fs = FileSystem.get(URI.create(ndUrl+path), conf);
		Path hdfsPath = new Path(ndUrl+path);
		@SuppressWarnings({ "deprecation" })
		SequenceFile.Reader reader = new SequenceFile.Reader(fs, hdfsPath, conf);
		Text key = new Text();
		HdpRenderedImageWritable val = new HdpRenderedImageWritable();
		reader.next(key, val);
		IOUtils.closeStream(reader);
		return val;
		
	}
	
	
	public boolean mkdirs(String path) throws IOException
	{
		Configuration conf = new Configuration();

		FileSystem fs = FileSystem.get(URI.create(ndUrl+path), conf);
		Path hdfsPath = new Path(ndUrl+path);
		return fs.mkdirs(hdfsPath);
	}
	
	
	public boolean isExist(String path) throws IOException
	{
		Configuration conf = new Configuration();

		FileSystem fs = FileSystem.get(URI.create(ndUrl+path), conf);
		Path hdfsPath = new Path(ndUrl+path);
		return fs.exists(hdfsPath);
	}
	
	
	public boolean delete(String path) throws IOException
	{
		Configuration conf = new Configuration();
		FileSystem fs = FileSystem.get(URI.create(ndUrl+path), conf);
		Path hdfsPath = new Path(ndUrl+path);
		return fs.delete(hdfsPath, true);
	}
	
	public boolean rename(String oldpath,String newpath) throws IOException
	{
		Configuration conf = new Configuration();
		FileSystem fs = FileSystem.get(conf);
		Path op=new Path(ndUrl+oldpath);
		Path np=new Path(ndUrl+newpath);
		return fs.rename(op, np);
	}
	
	public void downloadFromHDFS(String path,OutputStream out,int bufSize) throws IOException
	{
		Configuration conf = new Configuration();
		FileSystem fs = FileSystem.get(URI.create(ndUrl+path), conf);
		Path hdfsPath = new Path(ndUrl+path);
		FSDataInputStream fsdis = fs.open(hdfsPath);
		SendData.send(fsdis,out,bufSize);
		fsdis.close();
	}
	
	public long getLength(String path) throws IOException
	{
		Configuration conf = new Configuration();
		FileSystem fs = FileSystem.get(URI.create(ndUrl+path), conf);
		Path hdfsPath = new Path(ndUrl+path);
		return fs.getContentSummary(hdfsPath).getLength();
	}
	
	
	
}
